package org.ws.task;

import io.netty.util.concurrent.DefaultThreadFactory;
import org.apache.commons.collections4.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.ws.task.netty.handler.ActionHandlerRegistry;
import org.ws.task.netty.handler.DefaultActionHandlerRegistry;
import org.ws.task.netty.handler.PushItemWorkerHandler;
import org.ws.task.item.AbstractTask;
import org.ws.task.item.RemoteAccessApi;
import org.ws.task.item.TaskInstance;
import org.ws.task.repository.TaskRepository;
import org.ws.task.netty.NettyClient;
import org.ws.task.netty.NettyConfig;

import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class WorkerController implements Bootable {

    private static Logger log = LoggerFactory.getLogger(WorkerController.class.getName());

    private final NettyClient nettyClient;
    private final ActionHandlerRegistry actionHandlerRegistry;
    private final ExecutorService actionHandlerExecutor;
    private final ExecutorService callbackExecutor;
    private final ExecutorService processExecutor;
    private final PushItemWorkerHandler pushItemWorkerHandler;

    public WorkerController(NettyClient nettyClient, NettyConfig nettyConfig) {
        this.nettyClient = nettyClient;

        this.callbackExecutor = Executors.newFixedThreadPool(nettyConfig.getExecutorThreadNum(),
                new DefaultThreadFactory("WorkerCallbackThread"));


        this.actionHandlerExecutor = Executors.newFixedThreadPool(nettyConfig.getExecutorThreadNum(),
                new DefaultThreadFactory("WorkerActionHandlerThread"));

        this.processExecutor = Executors.newFixedThreadPool(nettyConfig.getExecutorThreadNum(),
                new DefaultThreadFactory("WorkerTaskProcessThread"));

        this.actionHandlerRegistry = new DefaultActionHandlerRegistry();

        this.pushItemWorkerHandler = new PushItemWorkerHandler(this.processExecutor);
    }

    @Override
    public void start(){
        Set<String> taskNames = TaskRepository.getInstance().taskNames();
        if(CollectionUtils.isNotEmpty(taskNames)){
            for (String name : taskNames) {
                TaskInstance taskInstance = TaskRepository.getInstance().getTaskInstance(name);
                if(taskInstance != null){
                    AbstractTask task = (AbstractTask)taskInstance.getTask();
                    task.start(new RemoteAccessApi(this.nettyClient,this.callbackExecutor));
                }
                else {
                    log.error("task[{}] can not find taskInstance.",name);
                }
            }
        }
        // action handler
        this.actionHandlerRegistry.register(this.pushItemWorkerHandler);
        this.nettyClient.actionHandlerRegistry(this.actionHandlerRegistry,this.actionHandlerExecutor);
        log.info("Worker Controller start.");
    }

    @Override
    public void shutdown() {
        this.callbackExecutor.shutdown();
        this.actionHandlerExecutor.shutdown();
        this.processExecutor.shutdown();
        this.actionHandlerRegistry.clear();
        this.nettyClient.shutdown();
    }

}
